package com.flink.es;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceFunctionProvider;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;

/**
 * @Author hujf
 * @Date 2022/8/29 10:20
 * @Description TODO
 */
public class HttpDynamicTableSource implements ScanTableSource {


    private String type;
    private String url;
    private String header;
    private String body;

    private DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
    private DataType producedDataType;

    private String unionType;
    private String unionUrl;
    private String unionData;

    public HttpDynamicTableSource(String type, String url, String header, String body, DecodingFormat<DeserializationSchema<RowData>> decodingFormat, DataType producedDataType, String unionType, String unionUrl, String unionData) {
        this.type = type;
        this.url = url;
        this.header = header;
        this.body = body;
        this.decodingFormat = decodingFormat;
        this.producedDataType = producedDataType;

        this.unionType = unionType;
        this.unionUrl = unionUrl;
        this.unionData = unionData;
    }

    @Override
    public ChangelogMode getChangelogMode() {
        return decodingFormat.getChangelogMode();
    }

    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
        final DeserializationSchema<RowData> deserializer = decodingFormat.createRuntimeDecoder(
                scanContext,
                producedDataType);
        HttpSourceFunction.Builder builder = HttpSourceFunction.builder()
                .setUrl(this.url).setBody(this.body).setUnionData(this.unionData)
                .setUnionUrl(this.unionUrl).setUnionType(this.unionType)
                .setType(this.type).setDeserializer(deserializer);

        return SourceFunctionProvider.of(builder.build(), true);
    }


    @Override
    public DynamicTableSource copy() {
        return new HttpDynamicTableSource(this.url, this.type, this.header, this.body, this.decodingFormat, this.producedDataType, unionType, unionUrl, unionData);
    }

    @Override
    public String asSummaryString() {
        return "HTTP Table Source";
    }
}
